package org.example.append_table;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 案例介绍：
 * 1  使用datagen生成日志数据
 * 2. 将生成的日志数据插入Paimon表
 * 3. 启动流任务
 */
public class LogWriter {
    public static void main(String[] args) throws Exception {
        // 配置Flink Web UI端口和并行度
        Configuration config = new Configuration();
        config.setString("rest.bind-port", "8081"); // 设置Web UI端口为8081
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        env.setParallelism(1); // 设置并行度为1
        env.enableCheckpointing(5000); // 设置checkpoint interval为5秒

        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 创建Paimon目录
        String createCatalogSQL = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");";
        System.out.println("正在执行SQL: \n" + createCatalogSQL);
        tableEnv.executeSql(createCatalogSQL);

        String dropLogTableSQL = "DROP TABLE IF EXISTS `paimon`.`default`.`LogTable`;";
        System.out.println("正在执行SQL: \n" + dropLogTableSQL);
        tableEnv.executeSql(dropLogTableSQL);

        // 创建日志表
        String createLogTableSQL = "CREATE TABLE IF NOT EXISTS `paimon`.`default`.`LogTable` (\n" +
                "    log_id STRING,\n" +
                "    log_time TIMESTAMP(3),\n" +
                "    level STRING,\n" +
                "    message STRING\n" +
                ") WITH (\n" +
                "    'connector' = 'paimon'\n" +
                "    , 'merge-engine' = 'partial-update'\n" +
                ");";
        System.out.println("正在执行SQL: \n" + createLogTableSQL);
        tableEnv.executeSql(createLogTableSQL);

        // 使用datagen生成日志数据
        String createTempTableSQL = "CREATE TEMPORARY TABLE TempLogTable (\n" +
                "    log_id STRING,\n" +
                "    log_time TIMESTAMP(3),\n" +
                "    level String,\n" +
                "    message STRING\n" +
                ") WITH (\n" +
                "    'connector' = 'datagen',\n" +
                "    'rows-per-second' = '10',\n" +
                "    'fields.log_id.length' = '10',\n" +
                "    'fields.level.length' = '5',\n" +
                "    'fields.message.length' = '20'\n" +
                ");";
        System.out.println("正在执行SQL: \n" + createTempTableSQL);
        tableEnv.executeSql(createTempTableSQL);

        // 将生成的日志数据插入Paimon表
        String insertSQL = "INSERT INTO `paimon`.`default`.`LogTable`\n" +
                "SELECT\n" +
                "    log_id,\n" +
                "    log_time,\n" +
                "    CASE WHEN RAND() < 0.2 THEN 'ERROR' WHEN RAND() < 0.5 THEN 'WARN' ELSE 'INFO' END AS level,\n" +
                "    message\n" +
                "FROM TempLogTable;";
        System.out.println("正在执行SQL: \n" + insertSQL);
        tableEnv.executeSql(insertSQL);
    }
}